AWS Parameters and Secrets Lambda ExtensionをPythonのrequestsモジュールなしで使ってみた
こんにちは、つくぼし(tsukuboshi0755)です!
LambdaでSSM Parameter StoreやSecrets Managerから値を取得する際に、AWS Parameters and Secrets Lambda Extension(以降Lambda Extension)を使うと、自前で実装しなくともキャッシュを利用でき、コストの削減やレイテンシーの改善を実現できます。
本機能をPythonで用いる場合、上記含め多くの記事でrequestsモジュールを用いたコードが記載されています。
requestsモジュールはコード量が少ないという特徴がある一方で、Python3.8以降では外部ライブラリになってしまっているため、使用する場合は別途レイヤーを作成する必要があります。これはちょっと面倒ですよね..。
そこで、今回はrequestsモジュールの代わりに、Python3.8以降も標準ライブラリに存在するurllib.requestモジュールを用いて、Lambda Extensionを使ってみたいと思います!
コード内容
以下のコードを用いる事で、SSM Parameter Store(String/Secure String)及びSecrets Managerから値を取得できます。
import urllib.request import os import json # Lambda関数ハンドラー def lambda_handler(event, context): string = get_string(os.environ.get('PARAM_NAME')) secure_string = get_secure_string(os.environ.get('SECURE_PARAM_NAME')) secret = get_secret(os.environ.get('SECRET_NAME'), os.environ.get('SECRET_KEY')) # 取得した情報を用いて、以下にLambdaで実施したい処理内容を記載 ... # Parameter StoreからStringを取得する関数 def get_string(param_name): params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + param_name headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')} params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers) with urllib.request.urlopen(params_extension_req) as response: param_config = response.read() param_value = json.loads(param_config)['Parameter']['Value'] return param_value # Parameter StoreからSecure Stringを取得する関数 def get_secure_string(secure_param_name): params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + secure_param_name + "&withDecryption=true" headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')} params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers) with urllib.request.urlopen(params_extension_req) as response: secure_param_config = response.read() secure_param_value = json.loads(secure_param_config.decode("utf-8"))['Parameter']['Value'] return secure_param_value # Secret Managerから取得する関数 def get_secret(secret_name, secret_key): secrets_extension_endpoint = "http://localhost:2773/secretsmanager/get?secretId=" + secret_name headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')} secrets_extension_req = urllib.request.Request(secrets_extension_endpoint, headers=headers) with urllib.request.urlopen(secrets_extension_req) as response: secret_config = response.read() secret_json = json.loads(secret_config)['SecretString'] secret_value = json.loads(secret_json)[secret_key] return secret_value
今回のコードではリクエストを送信する際に、requestsモジュールの代わりに、以下のurllib.requestモジュールを用いたリソース取得方法を使用しています。
urllib パッケージを使ってインターネット上のリソースを取得するには — Python 3.11.4 ドキュメント
上記のコードを用いる事で、現在LambdaでサポートされているPython3.7~3.11の全てで、追加のレイヤーを作成する事なくLambda Extensionを利用できます。
やってみた
実際に上記のコードで、Lambda ExtensionからSSM Parameter Store及びSecrets Managerの値を取得できるか確認してみます。
コードの動作検証をするため、適当な値を、Parameter Store及びSecret Managerに保存します。
今回は以下の内容で、3つのサービスに分けて値を保存し、Lambdaで取り出せるか試します。
サービス | 種類 | 名前 | キー | 値 |
---|---|---|---|---|
Parameter Store | String | /cm-tsukuboshi/param | - | hoge |
Parameter Store | Secure String | /cm-tsukuboshi/secure-param | - | huga |
Secrets Manager | - | /cm-tsukuboshi/secret | sample | piyo |
まず以下の通り、Lambda関数をPythonで作成します。
次に以下の通り、Lambda関数のレイヤーにLambda Extensionを追加します。
続いて、Lambdaの環境変数を以下の通り設定します。
キー | 値 | 補足 |
---|---|---|
PARAM_NAME | /cm-tsukuboshi/param | SSM Parameter Storeに保存したパラメータ(String)の名前 |
SECURE_PARAM_NAME | /cm-tsukuboshi/secure-param | SSM Parameter Storeに保存したパラメータ(Secure String)の名前 |
SECRET_NAME | /cm-tsukuboshi/secret | Secrets Managerに保存したシークレットの名前 |
SECRET_KEY | sample | Secrets Managerに保存したシークレットのキー |
さらにLambda実行ロールに対して、SSM Parameter Store及びSecrets Managerへのアクセス権が付与されたポリシーを作成しアタッチします。
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "ssm:GetParameter*" ], "Resource": [ "arn:aws:secretsmanager:<region>:<account-id>:parameter:<parameter-name>" ] }, { "Effect": "Allow", "Action": [ "ssm:GetParameter*" ], "Resource": [ "arn:aws:secretsmanager:<region>:<account-id>:parameter:<secure-parameter-name>" ] }, { "Effect": "Allow", "Action": [ "kms:Decrypt" ], "Resource": [ "arn:aws:kms:<region>:<account-id>:key/<default-ssm-key-id>" ] }, { "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": [ "arn:aws:secretsmanager:<region>:<account-id>:secret:<secret-name>" ] } ] }
そしてLambdaコードを以下の内容でデプロイします。
import urllib.request import os import json # Lambda関数ハンドラー def lambda_handler(event, context): string = get_string(os.environ.get('PARAM_NAME')) secure_string = get_secure_string(os.environ.get('SECURE_PARAM_NAME')) secret = get_secret(os.environ.get('SECRET_NAME'), os.environ.get('SECRET_KEY')) # CloudWatch Logsに取得した内容を出力 print("From SSM Parameter Store (String): " + string) print("From SSM Parameter Store (SecureString): " + secure_string) print("From Secret Manager: " + secret) # Parameter StoreからStringを取得する関数 def get_string(param_name): params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + param_name headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')} params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers) with urllib.request.urlopen(params_extension_req) as response: param_config = response.read() param_value = json.loads(param_config)['Parameter']['Value'] return param_value # Parameter StoreからSecure Stringを取得する関数 def get_secure_string(secure_param_name): params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + secure_param_name + "&withDecryption=true" headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')} params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers) with urllib.request.urlopen(params_extension_req) as response: secure_param_config = response.read() secure_param_value = json.loads(secure_param_config.decode("utf-8"))['Parameter']['Value'] return secure_param_value # Secret Managerから取得する関数 def get_secret(secret_name, secret_key): secrets_extension_endpoint = "http://localhost:2773/secretsmanager/get?secretId=" + secret_name headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')} secrets_extension_req = urllib.request.Request(secrets_extension_endpoint, headers=headers) with urllib.request.urlopen(secrets_extension_req) as response: secret_config = response.read() secret_json = json.loads(secret_config)['SecretString'] secret_value = json.loads(secret_json)[secret_key] return secret_value
なお今回のコードは動作確認が目的のため、一目で分かるようにprint関数を用いて、CloudWatch Logsに保存した値を出力しています。
本番環境で上記コードを使用する場合、基本的に取得した秘密情報はCloudWatch Logsに出力しないでください。
以上が完了した後、Lambdaのテストボタンで関数を実行し、CloudWatch Logsで結果を確認します。
SSM Parameter Store及びSecrets Managerから、hoge
,fuga
,piyo
の値が取得できている事が分かりますね。
最後に
今回はrequestsモジュールの代わりにurllib.requestモジュールを用いて、Lambda Extensionを使ってみました。
requestsモジュールと比較して少し記述量は多くなってしまいますが、現状LambdaがサポートしているPythonの全てのバージョンで標準ライブラリとして使用できるのは嬉しいですね。
Lambda Extensionは便利な機能ですので、Lambdaを書く機会がある方は活用を検討してみると良いかもしれません。
以上、つくぼし(tsukuboshi0755)でした!